package com.atguigu.flink.chapter04_window.agg;

import com.atguigu.flink.pojo.MyUtil;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/10/25
 *
 * Reduce: 滚动聚合
 *      通过过去每种传感器，每3个的vc总和
 *              两两聚合，聚合的参数和返回值必须是同一种类型，
 *
 *
 *               public <R> SingleOutputStreamOperator<R> reduce(
 *             ReduceFunction<T> reduceFunction, //聚合函数。 每滚动一次执行一次
 *             WindowFunction<T, R, K, W(窗口)> function  //有窗口信息，在窗口被触发时，才执行)
 *
 */
public class Demo12_ReduceTime
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .keyBy(WaterSensor::getId)
           .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
           .reduce(new ReduceFunction<WaterSensor>()
           {
               @Override
               public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                   System.out.println("Demo11_Reduce.reduce");
                   //value2.setVc(value1.getVc() + value2.getVc());
                   value1.setVc(value1.getVc() + value2.getVc());
                   return value1;
               }
           }, new WindowFunction<WaterSensor, String, String, TimeWindow>()
           {
               /*
                    在窗口触发时，打印下到底聚合了哪些数据

                    apply： 全量聚合。等窗口全凑齐，被触发了，执行一次。
                            和process一样

                    Iterable<WaterSensor> input: 滚动聚合，reduce最终只有1个元素，最后reduce的结果

                    只有reduce，reduce把最终的结果，发到print
                    有reduce + apply，reduce把结果发给apply，由apply再发往print
                */
               @Override
               public void apply(String s, TimeWindow window, Iterable<WaterSensor> input, Collector<String> out) throws Exception {

                   MyUtil.printTimeWindow(window);
                   out.collect(MyUtil.toList(input).toString());

               }
           })
           .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
